package com.gs.netty.client.remoting;


import com.gs.netty.client.*;
import com.gs.netty.client.enums.CommunicationMode;
import com.gs.netty.client.exception.*;
import com.gs.netty.client.util.ChannelUtils;

import com.gs.netty.core.SemaphoreReleaseOnlyOnce;
import com.gs.netty.support.server.AbstractServer;

import com.gs.netty.websocket.remoting.WebSocketInvocation;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;
import javax.net.ssl.SSLContext;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;

@Slf4j
public abstract class AbstractRemotingClient extends AbstractServer implements RemotingClient {


    public AbstractRemotingClient(NettyClientConfig nettyClientConfig) throws Exception {
        super(nettyClientConfig);
    }

    @Override
    public <T> T send(Message message, Class<T> requiredType) throws WebSocketClientHandshakerException, WebSocketRemotingConnectException, WebSocketNotFountChannelException, WebSocketTooMuchRequestTimeoutException, WebSocketClientException {
        return send(message,getSendTimeout(),requiredType);
    }

    @Override
    public SendResult send(Message message) throws WebSocketClientHandshakerException, WebSocketRemotingConnectException, WebSocketNotFountChannelException, WebSocketTooMuchRequestTimeoutException, WebSocketClientException {
        return send(message,getSendTimeout(),null);
    }



    @Override
    public SendResult send(Message message, long timeoutMillis) throws WebSocketClientHandshakerException, WebSocketRemotingConnectException, WebSocketNotFountChannelException, WebSocketTooMuchRequestTimeoutException, WebSocketClientException {
        return send(message,timeoutMillis,SendResult.class);
    }

    @Override
    public <T> T send(Message message, long timeoutMillis, Class<T> requiredType) throws WebSocketClientHandshakerException, WebSocketRemotingConnectException, WebSocketNotFountChannelException, WebSocketTooMuchRequestTimeoutException, WebSocketClientException {

        return (T) doSend(message,CommunicationMode.SYNC,null,timeoutMillis,requiredType);

    }

    @Override
    public void send(Message commandRequest, SendCallback sendCallback) throws WebSocketClientHandshakerException, WebSocketRemotingConnectException, WebSocketNotFountChannelException, WebSocketTooMuchRequestTimeoutException, WebSocketClientException {
         send(commandRequest,sendCallback,getSendTimeout());
    }

    @Override
    public void send(Message commandRequest, SendCallback sendCallback, long timeoutMillis) throws WebSocketClientHandshakerException, WebSocketRemotingConnectException, WebSocketNotFountChannelException, WebSocketTooMuchRequestTimeoutException, WebSocketClientException {
        doSend(commandRequest,CommunicationMode.ASYNC,sendCallback,timeoutMillis,null);
    }

    protected abstract  <T> Object doSend(Message message, CommunicationMode mode,
                                          SendCallback sendCallback, long timeout, Class<T> requiredType)

            throws WebSocketTooMuchRequestTimeoutException, WebSocketClientHandshakerException, WebSocketRemotingConnectException, WebSocketNotFountChannelException, WebSocketClientException;




    protected WebSocketInvocation invokeSync0(final Channel channel,
                                      final WebSocketInvocation rpcInvocation,
                                      final long timeoutMillis) throws WebSocketRemotingTimeoutException,
            WebSocketRemotingSendRequestException {
        try {

            // 远程调用
            return doInvoke(channel, rpcInvocation,null, timeoutMillis)
                            .thenApply(ResponseFuture::getResponseCommand)
                    .get(timeoutMillis, TimeUnit.MILLISECONDS);
        } catch (ExecutionException  | InterruptedException e) {
            throw new WebSocketRemotingSendRequestException(channel.remoteAddress().toString(), e.getCause());
        } catch (TimeoutException e) {
            throw new WebSocketRemotingTimeoutException(channel.remoteAddress().toString(), timeoutMillis, e.getCause());
        }
    }

    protected abstract   Semaphore getSemaphoreAsync();

    private CompletableFuture<ResponseFuture> doInvoke(Channel channel,
                                                       WebSocketInvocation rpcInvocation,
                                                              SendCallback sendCallback,
                                                              long timeoutMillis) {

        long beginStartTime = System.currentTimeMillis();
        // 结果响应
        CompletableFuture<ResponseFuture> future = new CompletableFuture<>();
        // 设置请求Id
        Integer requestId = rpcInvocation.getRequestId();
        boolean acquired;
        try {
            // 获取一个发送信号 这个确保了客户端只能发送多少给异步请求防止用户项目宕机
            acquired = getSemaphoreAsync().tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        } catch (Throwable t) {
            future.completeExceptionally(t);
            return future;
        }

        if (acquired) {

            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(getSemaphoreAsync());
            long costTime = System.currentTimeMillis() - beginStartTime;
            if (timeoutMillis < costTime) {
                once.release();
                future.completeExceptionally(new WebSocketRemotingTimeoutException("invokeAsyncImpl call timeout"));
                return future;
            }

            //封装ResponseFuture，确保 ResponseFuture 对象值线程可见性 底层就是 volatile 关键字
            AtomicReference<ResponseFuture> responseFutureReference = new AtomicReference<>();


            // 创建响应
            ResponseFuture responseFuture = new ResponseFuture(channel, requestId,rpcInvocation,
                    timeoutMillis, new InvokeCallback() {
                @Override
                public void complete(ResponseFuture responseFuture) {
                }
                @Override
                public void succeed(WebSocketInvocation response) {
                    // 设置响应结果 ，唤醒  future 等线程
                    future.complete(responseFutureReference.get());
                }
                @Override
                public void fail(Throwable throwable) {
                    // 设置异常结果 ，唤醒  future 等线程
                    future.completeExceptionally(throwable);
                }
            },once);

            // 设置对象可见性
            responseFutureReference.set(responseFuture);

            // 添加响应表
            getResponseTables().put(requestId, responseFuture);

            try {
                // 发送数据
                channel.writeAndFlush(rpcInvocation).addListener((ChannelFutureListener) f -> {
                    // 判断是否发送成功
                    if (f.isSuccess()) {
                        responseFuture.setSendRequestOK(true);
                        return;
                    }
                    // 发送失败
                    requestFail(requestId);
                    String addr = ChannelUtils.parseChannelRemoteAddr(channel);
                    log.warn("send a request command to channel <{}> failed.", addr);
                });
                return future;
            } catch (Exception e) { // 发送异常
                // 移除响应表
                getResponseTables().remove(requestId);
                String addr = ChannelUtils.parseChannelRemoteAddr(channel);
                log.warn("send a request command to channel <" + addr + "> Exception", e);
                future.completeExceptionally(new WebSocketRemotingSendRequestException(addr, e));
                return future;
            }
        }else {
            if (timeoutMillis <= 0) {
                future.completeExceptionally(new WebSocketTooMuchRequestTimeoutException("invokeAsyncImpl invoke too fast"));
            } else {
                String info =
                        String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                                timeoutMillis,
                                getSemaphoreAsync().getQueueLength(),
                                getSemaphoreAsync().availablePermits()
                        );
                log.warn(info);
                future.completeExceptionally(new WebSocketRemotingTimeoutException(info));
            }
            return future;
        }
    }

    protected abstract void requestFail(Integer requestId);


    /**
     * 获取请求响应表
     * @return
     */
    protected  abstract  ConcurrentHashMap<Integer, ResponseFuture> getResponseTables();

    /**
     * 获取发送消息超时时间
     * @return
     */
    protected abstract long getSendTimeout();
}
